package cn.z2huo.knowledge.concurrency.queue;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.RandomUtil;
import lombok.Getter;
import lombok.RequiredArgsConstructor;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.*;

/**
 * <p>阻塞队列，生产者-消费者 模式示例
 *
 * @author z2huo
 */
class BlockingQueueProducerConsumerTest {

    private static final int QUEUE_CAPACITY = 5;

    @Getter
    private final BlockingQueue<String> queue = new ArrayBlockingQueue<>(QUEUE_CAPACITY);

    @RequiredArgsConstructor
    static class Producer implements Runnable {

        private final BlockingQueue<String> queue;

        @Override
        public void run() {
            try {
                while (true) {
                    String item = this.produceItem();
                    // 如果队列满，则阻塞等待
                    queue.put(item);
                    System.out.println("thread is " + Thread.currentThread().getName() + ", Produced: " + item);
                    ThreadUtil.sleep(1, TimeUnit.SECONDS);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

        private String produceItem() {
            return LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME) + " " + RandomUtil.randomString(8);
        }
    }

    @RequiredArgsConstructor
    static class Consumer implements Runnable {

        private final BlockingQueue<String> queue;

        @Override
        public void run() {
            try {
                while (true) {
                    // 如果队列空，则阻塞等待
                    String item = queue.take();
                    // 消费物品
                    this.consumeItem(item);
                    System.out.println("thread is " + Thread.currentThread().getName() + ", Consumed: " + item);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

        private void consumeItem(String item) {
            //System.out.println("Consuming: " + item);
        }
    }

    public static void main(String[] args) {

        BlockingQueueProducerConsumerTest test = new BlockingQueueProducerConsumerTest();

        try (ExecutorService executorService = Executors.newFixedThreadPool(10)) {
            // 阻塞队列容量为 5，如果启用大于阻塞队列容量的生产者，则队列已满，将会阻塞等待直到队列有空余容量为止
            for (int i = 0; i < QUEUE_CAPACITY; i++) {
                executorService.execute(new Producer(test.getQueue()));
            }

            for (int i = 0; i < 2; i++) {
                executorService.execute(new Consumer(test.getQueue()));
            }
        }
    }

}
